﻿//
// (C) Copyright 2009 Irantha Suwandarathna (iranthas@hotmail.com)
// All rights reserved.
//

/* Copyright (c) 2001-2008, The HSQL Development Group
 * All rights reserved.
 *
 * Redistribution and use _in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 *
 * Redistributions _in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer _in the documentation
 * and/or other materials provided with the distribution.
 *
 * Neither the name of the HSQL Development Group nor the names of its
 * contributors may be used to endorse or promote products derived from this
 * software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL HSQL DEVELOPMENT GROUP, HSQLDB.ORG,
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using EffiProz.Core.Lib;
using EffiProz.Core.Persist;
using EffiProz.Core.RowIO;
using EffiProz.Core.Indexes;
using EffiProz.Core.Errors;
using EffiProz.Core.Rights;
using EffiProz.Core.Navigators;
using EffiProz.Core.DataTypes;
using EffiProz.Core.Results;

namespace EffiProz.Core
{
    public class TransactionManagerMVCC : TransactionManager
    {
        Database database;
        bool mvcc = false;

#if !SILVERLIGHT
        ReaderWriterLockSlim _lock = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
#else
        private class Dummy { };
        Object _lock = new Dummy();
#endif
 
        // functional unit - sessions involved in live transactions

        /** live transactions keeping committed transactions from being merged */
        public LongDeque liveTransactionTimestamps = new LongDeque();
#if SILVERLIGHT
        private  class Dummy2 { };
        private Dummy2 globalChangeTimestamp_synch = new Dummy2();
#endif
        public long globalChangeTimestamp = 0;

        // functional unit - merged committed transactions
        EfzDeque committedTransactions = new EfzDeque();
        LongDeque committedTransactionTimestamps = new LongDeque();

        // functional unit - cached table transactions

        /** Map : rowID -> RowAction */
        public IntKeyHashMapConcurrent rowActionMap =
            new IntKeyHashMapConcurrent(10000);

        //
        //
        HashMap tableWriteLocks = new HashMap();
        MultiValueHashMap tableReadLocks = new MultiValueHashMap();

        public TransactionManagerMVCC(Database db)
        {
            database = db;
            mvcc = true;
        }

        public long getGlobalChangeTimestamp()
        {   
#if !SILVERLIGHT            
            return Interlocked.Read(ref globalChangeTimestamp);
#else
            long tmp;
            lock(globalChangeTimestamp_synch)
            {
                tmp = globalChangeTimestamp;
            }
            return tmp;
#endif
        }

        public bool isMVRows()
        {
            return true;
        }

        public int getTransactionControl()
        {
            return mvcc ? Database.MVCC
                        : Database.MVLOCKS;
        }

        public void setTransactionControl(Session session, int mode)
        {

#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
            Monitor.Enter(_lock);
#endif

            try
            {
                lock (liveTransactionTimestamps)
                {

                    // statement runs as transaction
                    if (liveTransactionTimestamps.size() == 1)
                    {
                        switch (mode)
                        {

                            case Database.MVCC:
                                mvcc = true;
                                break;
#if !SILVERLIGHT
                            case Database.MVLOCKS:
                                {
                                    TransactionManagerMV2PL manager =
                                        new TransactionManagerMV2PL(database);

                                    long tmp = Interlocked.Read(ref globalChangeTimestamp);
                                    Interlocked.Exchange(ref manager.globalChangeTimestamp, tmp);
                                    manager.liveTransactionTimestamps.addLast(
                                        session.transactionTimestamp);

                                    database.txManager = manager;

                                    break;
                                }
                            case Database.LOCKS:
                                {
                                    TransactionManager2PL manager =
                                        new TransactionManager2PL(database);

                                    long tmp = Interlocked.Read(ref globalChangeTimestamp);
                                    Interlocked.Exchange(ref manager.globalChangeTimestamp, tmp);

                                    database.txManager = manager;
                                    break;
                                }
#endif
                        }

                        return;
                    }
                }
            }
            finally
            {
  
#if !SILVERLIGHT
            _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
            }

            throw Error.error(ErrorCode.X_25001);
        }

        public void completeActions(Session session)
        {

            Object[] list = session.rowActionList.getArray();
            int limit = session.rowActionList.size();
            bool canComplete = true;

#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
            Monitor.Enter(_lock);
#endif

            try
            {
                for (int i = session.actionIndex; i < limit; i++)
                {
                    RowAction rowact = (RowAction)list[i];

                    if (rowact.complete(session, session.tempSet))
                    {
                        continue;
                    }

                    canComplete = false;

                    if (session.isolationMode == SessionTxConsts
                            .TX_REPEATABLE_READ || session
                            .isolationMode == SessionTxConsts.TX_SERIALIZABLE)
                    {
                        session.abortTransaction = true;

                        break;
                    }
                }

                for (int i = session.actionIndex; canComplete && i < limit; i++)
                {
                    RowAction action = (RowAction)list[i];

                    if (!action.table.isLogged)
                    {
                        continue;
                    }

                    Row row = action.memoryRow;

                    if (row == null)
                    {
                        PersistentStore store =
                            session.sessionData.getRowStore(action.table);

                        row = (Row)store.get(action.getPos(), false);
                    }

                    Object[] data = row.getData();

                    try
                    {
                        int actionType =
                            action.getActionType(session.actionTimestamp);

                        if (actionType == RowActionBase.ACTION_INSERT)
                        {
                            database.logger.writeInsertStatement(session,
                                                                 action.table,
                                                                 data);
                        }
                        else if (actionType == RowActionBase.ACTION_DELETE)
                        {
                            database.logger.writeDeleteStatement(session,
                                                                 action.table,
                                                                 data);
                        }
                        else if (actionType == RowActionBase.ACTION_NONE)
                        {

                            // no logging
                        }
                        else
                        {
                            throw Error.runtimeError(ErrorCode.U_S0500,
                                                     "TransactionManager");
                        }
                    }
                    catch (CoreException )
                    {

                        // can put db in special state
                    }
                }

                if (!canComplete && !session.abortTransaction)
                {
                    session.redoAction = true;

                    rollbackAction(session);

                    if (!session.tempSet.isEmpty())
                    {
                        session.latch.setCount(session.tempSet.size());

                        for (int i = 0; i < session.tempSet.size(); i++)
                        {
                            Session current = (Session)session.tempSet.get(i);

                            current.waitingSessions.add(session);

                            // waitedSessions.put(current, session);
                            // waitingSessions.put(session, current);
                        }
                    }
                }
            }
            finally
            {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
                session.tempSet.clear();
            }
        }

        public bool prepareCommitActions(Session session)
        {

            Object[] list = session.rowActionList.getArray();
            int limit = session.rowActionList.size();

            if (session.abortTransaction)
            {

                //            System.out.println("cascade fail " + session + " " + session.actionTimestamp);
                return false;
            }

            try
            {
#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
                Monitor.Enter(_lock);
#endif

                for (int i = 0; i < limit; i++)
                {
                    RowAction rowact = (RowAction)list[i];

                    if (!rowact.canCommit(session, session.tempSet))
                    {

                        //                System.out.println("commit conflicts " + session + " " + session.actionTimestamp);
                        return false;
                    }
                }

                session.actionTimestamp = nextChangeTimestamp();

                for (int i = 0; i < limit; i++)
                {
                    RowAction action = (RowAction)list[i];

                    action.prepareCommit(session);
                }

                for (int i = 0; i < session.tempSet.size(); i++)
                {
                    Session current = (Session)session.tempSet.get(i);

                    current.abortTransaction = true;
                }

                return true;
            }
            finally
            {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
                session.tempSet.clear();
            }
        }

        public bool commitTransaction(Session session)
        {

            if (session.abortTransaction)
            {
                return false;
            }

            int limit = session.rowActionList.size();
            Object[] list = limit == 0 ? new object[0]
                                        : session.rowActionList.getArray();

            try
            {
#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
                Monitor.Enter(_lock);
#endif

                for (int i = 0; i < limit; i++)
                {
                    RowAction rowact = (RowAction)list[i];

                    if (!rowact.canCommit(session, session.tempSet))
                    {

                        //                System.out.println("commit conflicts " + session + " " + session.actionTimestamp);
                        return false;
                    }
                }

                endTransaction(session);

                if (limit == 0)
                {
                    if (!mvcc)
                    {
                        endTransactionTPL(session);
                    }

                    try
                    {
                        session.logSequences();

                        if (mvcc)
                        {
                            database.logger.writeCommitStatement(session);
                        }
                    }
                    catch (CoreException ) { }

                    return true;
                }

                // new actionTimestamp used for commitTimestamp
                session.actionTimestamp = nextChangeTimestamp();

                for (int i = 0; i < limit; i++)
                {
                    RowAction action = (RowAction)list[i];

                    action.commit(session);
                }

                for (int i = 0; i < session.tempSet.size(); i++)
                {
                    Session current = (Session)session.tempSet.get(i);

                    current.abortTransaction = true;
                }

                for (int i = 0; i < limit; i++)
                {
                    RowAction action = (RowAction)list[i];

                    if (action.table.hasLobColumn)
                    {
                        int type = action.getCommitType(session.actionTimestamp);

                        switch (type)
                        {

                            case RowActionBase.ACTION_INSERT:
                                Row row = action.memoryRow;

                                if (row == null)
                                {
                                    PersistentStore store =
                                        session.sessionData.getRowStore(
                                            action.table);

                                    row = (Row)store.get(action.getPos(), false);
                                }

                                action.table.addLobUsageCount(session,
                                                              row.getData());
                                break;

                         
                        }
                    }

                    if (action.table.tableType == TableBase.TEXT_TABLE)
                    {
                        PersistentStore store =
                            session.sessionData.getRowStore(action.table);
                        int type = action.getCommitType(session.actionTimestamp);

                        switch (type)
                        {

                            case RowActionBase.ACTION_DELETE:
                                store.removePersistence(action.getPos());
                                break;

                            case RowActionBase.ACTION_INSERT:
                                Row row = (Row)store.get(action.getPos(), false);

                                store.commitPersistence(row);
                                break;

                        }
                    }
                }

                // session.actionTimestamp is the committed tx timestamp
                if (getFirstLiveTransactionTimestamp() > session.actionTimestamp)
                {
                    mergeTransaction(session, list, 0, limit,
                                     session.actionTimestamp);
                    rowActionMapRemoveTransaction(list, 0, limit, true);
                }
                else
                {
                    list = session.rowActionList.toArray();

                    addToCommittedQueue(session, list);
                }

                try
                {
                    session.logSequences();
                    database.logger.writeCommitStatement(session);
                }
                catch (CoreException ) { }

                //
                if (mvcc)
                {
                    countDownLatches(session);
                }
                else
                {
                    endTransactionTPL(session);
                }

                return true;
            }
            finally
            {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
                session.tempSet.clear();
            }
        }

        private void countDownLatches(Session session)
        {

            if (!session.waitingSessions.isEmpty())
            {
                for (int i = 0; i < session.waitingSessions.size(); i++)
                {
                    Session current = (Session)session.waitingSessions.get(i);

                    current.latch.countDown();
                }

                session.waitingSessions.clear();
            }
        }

        public void rollback(Session session)
        {

            session.abortTransaction = false;
            session.actionTimestamp = nextChangeTimestamp();

            rollbackPartial(session, 0, session.transactionTimestamp);
            endTransaction(session);


            if (mvcc)
            {
                //
            }
            else
            {
                try
                {

#if !SILVERLIGHT
                _lock.EnterWriteLock();
#else
                    Monitor.Enter(_lock);
#endif

                    endTransactionTPL(session);
                }
                finally
                {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                    Monitor.Exit(_lock);
#endif
                }
            }
        }

        public void rollbackSavepoint(Session session, int index)
        {

            long timestamp = session.sessionContext.savepointTimestamps.get(index);
            int oi = Convert.ToInt32(session.sessionContext.savepoints.get(index));
            int start = oi;

            while (session.sessionContext.savepoints.size() > index + 1)
            {
                session.sessionContext.savepoints.remove(
                    session.sessionContext.savepoints.size() - 1);
                session.sessionContext.savepointTimestamps.removeLast();
            }

            rollbackPartial(session, start, timestamp);
        }

        public void rollbackAction(Session session)
        {
            rollbackPartial(session, session.actionIndex, session.actionTimestamp);
        }

        /**
         * rollback the row actions from start index in list and
         * the given timestamp
         */
        public void rollbackPartial(Session session, int start, long timestamp)
        {

            Object[] list = session.rowActionList.getArray();
            int limit = session.rowActionList.size();

            if (start == limit)
            {
                return;
            }

            for (int i = start; i < limit; i++)
            {
                RowAction action = (RowAction)list[i];

                if (action != null)
                {
                    action.rollback(session, timestamp);
                }
                else
                {
                    System.Diagnostics.Debug.WriteLine("null action in rollback " + start);
                }
            }

            // rolled back transactions can always be merged as they have never been
            // seen by other sessions
            mergeRolledBackTransaction(session.rowActionList.getArray(), start,
                                       limit);
            rowActionMapRemoveTransaction(session.rowActionList.getArray(), start,
                                          limit, false);
            session.rowActionList.setSize(start);
        }

        public RowAction addDeleteAction(Session session, Table table, Row row)
        {

            RowAction action;

            lock (row)
            {
                action = RowAction.addAction(session, RowActionBase.ACTION_DELETE,
                                             table, row);
            }

            session.rowActionList.add(action);

            if (!row.isMemory())
            {
                rowActionMap.put(action.getPos(), action);
            }

            return action;
        }

        public void addInsertAction(Session session, Table table, Row row)
        {

            RowAction action = row.rowAction;

            if (action == null)
            {
                System.Diagnostics.Debug.WriteLine("null insert action " + session + " "
                                   + session.actionTimestamp);
            }

            session.rowActionList.add(action);

            if (!row.isMemory())
            {
                rowActionMap.put(action.getPos(), action);
            }
        }

        // functional unit - row comparison

        /**
         * Used for inserting rows into unique indexes
         */

        /** @todo test skips some equal rows because it relies on getPos() */
        public int compareRowForInsert(Row newRow, Row existingRow)
        {

            // only allow new inserts over rows deleted by the same session
            if (newRow.rowAction != null
                    && !canRead(newRow.rowAction.session, existingRow))
            {
                return newRow.getPos() - existingRow.getPos();
            }

            return 0;
        }

        // functional unit - accessibility of rows
        public bool canRead(Session session, Row row)
        {

            RowAction action = row.rowAction;

            if (action == null)
            {
                return true;
            }

            return action.canRead(session);
        }

        public bool isDeleted(Session session, Row row)
        {

            RowAction action = row.rowAction;

            if (action == null)
            {
                return false;
            }

            return !action.canRead(session);
        }

        public bool canRead(Session session, int id)
        {

            RowAction action = (RowAction)rowActionMap.get(id);

            return action == null ? true
                                  : action.canRead(session);
        }
        /**
         *  todo - can remove a row that has previously inserted by the same transaction
         */
        public void rowActionMapRemoveTransaction(Object[] list, int start, int limit,
                                           bool commit)
        {

            for (int i = start; i < limit; i++)
            {
                RowAction rowact = (RowAction)list[i];

                if (!rowact.isMemory)
                {
                    lock (rowact)
                    {
                        if (rowact.type == RowActionBase.ACTION_NONE
                                || rowact.type
                                   == RowActionBase.ACTION_DELETE_FINAL)
                        {
                            int pos = rowact.getPos();

                            rowActionMap.remove(pos);
                        }
                    }
                }
            }

            deleteRows(list, start, limit, commit);
        }

        public void deleteRows(Object[] list, int start, int limit, bool commit)
        {

            for (int i = start; i < limit; i++)
            {
                RowAction rowact = (RowAction)list[i];

                if (rowact.type == RowActionBase.ACTION_DELETE_FINAL)
                {
                    try
                    {
                        rowact.type = RowActionBase.ACTION_DELETE_COMMITTED;

                        PersistentStore store =
                            rowact.session.sessionData.getRowStore(rowact.table);
                        Row row = rowact.memoryRow;

                        if (row == null)
                        {
                            row = (Row)store.get(rowact.getPos(), false);
                        }

                        if (commit && rowact.table.hasLobColumn)
                        {
                            Object[] data = row.getData();

                            rowact.table.removeLobUsageCount(rowact.session, data);
                        }

                        store.delete(row);
                        store.remove(row.getPos());
                    }
                    catch (CoreException )
                    {

                        //                    throw unexpectedException(e.getMessage());
                    }
                }
            }
        }

        /**
         * add transaction info to a row just loaded from the cache. called only
         * for CACHED tables
         */
        public void setTransactionInfo(CachedObject obj)
        {

            Row row = (Row)obj;

            if (row.rowAction != null)
            {
                return;
            }

            RowAction rowact = (RowAction)rowActionMap.get(row.position);

            row.rowAction = rowact;
        }

        /**
         * merge a given list of transaction rollback action with given timestamp
         */
        public void mergeRolledBackTransaction(Object[] list, int start, int limit)
        {

            for (int i = start; i < limit; i++)
            {
                RowAction rowact = (RowAction)list[i];

                if (rowact == null || rowact.type == RowActionBase.ACTION_NONE
                        || rowact.type == RowActionBase.ACTION_DELETE_FINAL)
                {
                    continue;
                }

                Row row = rowact.memoryRow;

                if (row == null)
                {
                    Session session = rowact.session;

                    if (session == null)
                    {
                        return;
                    }

                    PersistentStore store =
                        rowact.session.sessionData.getRowStore(rowact.table);

                    row = (Row)store.get(rowact.getPos(), false);
                }

                if (row == null)
                {
                    continue;
                }

                lock (row)
                {
                    rowact.mergeRollback(row);
                }
            }

            //        } catch (Exception t) {
            //            System.out.println("throw in merge");
            //            t.printStackTrace();
            //        }
        }

        /**
         * add a list of actions to the end of queue
         */
        public void addToCommittedQueue(Session session, Object[] list)
        {

            lock (committedTransactionTimestamps)
            {

                // add the txList according to commit timestamp
                committedTransactions.addLast(list);

                // get session commit timestamp
                committedTransactionTimestamps.addLast(session.actionTimestamp);
                /* debug 190
                            if (committedTransactions.size() > 64) {
                                System.out.println("******* excessive transaction queue");
                            }
                // debug 190 */
            }
        }

        /**
         * expire all committed transactions that are no longer in scope
         */
        public void mergeExpiredTransactions(Session session)
        {

            long timestamp = getFirstLiveTransactionTimestamp();

            while (true)
            {
                long commitTimestamp = 0;
                Object[] actions = null;

                lock (committedTransactionTimestamps)
                {
                    if (committedTransactionTimestamps.isEmpty())
                    {
                        break;
                    }

                    commitTimestamp = committedTransactionTimestamps.getFirst();

                    if (commitTimestamp < timestamp)
                    {
                        committedTransactionTimestamps.removeFirst();

                        actions = (Object[])committedTransactions.removeFirst();
                    }
                    else
                    {
                        break;
                    }
                }

                mergeTransaction(session, actions, 0, actions.Length,
                                 commitTimestamp);
                rowActionMapRemoveTransaction(actions, 0, actions.Length, true);
            }
        }

        /**
         * merge a transaction committed at a given timestamp.
         */
        public void mergeTransaction(Session session, Object[] list, int start,
                              int limit, long timestamp)
        {

            for (int i = start; i < limit; i++)
            {
                RowAction rowact = (RowAction)list[i];

                if (rowact == null || rowact.type == RowActionBase.ACTION_NONE
                        || rowact.type == RowActionBase.ACTION_DELETE_FINAL)
                {
                    continue;
                }

                Row row = rowact.memoryRow;

                if (row == null)
                {
                    PersistentStore store =
                        rowact.session.sessionData.getRowStore(rowact.table);

                    row = (Row)store.get(rowact.getPos(), false);
                }

                if (row == null)
                {
                    continue;
                }

                lock (row)
                {
                    rowact.mergeToTimestamp(row, timestamp);
                }
            }
        }

        /**
         * gets the next timestamp for an action
         */
        public long nextChangeTimestamp()
        {

#if !SILVERLIGHT            
            return Interlocked.Increment(ref globalChangeTimestamp);
#else
            long tmp;
            lock (globalChangeTimestamp_synch)
            {
                tmp = ++globalChangeTimestamp;
            }
            return tmp;
#endif
        }

        public void beginTransaction(Session session)
        {

            lock (liveTransactionTimestamps)
            {
                session.actionTimestamp = nextChangeTimestamp();
                session.transactionTimestamp = session.actionTimestamp;
                session.isTransaction = true;

                liveTransactionTimestamps.addLast(session.transactionTimestamp);

                try
                {
                    if (mvcc)
                    {
                        database.logger.writeToLog(
                            session, session.getStartTransactionSQL());
                    }
                }
                catch (CoreException ) { }
            }
        }

        /**
         * add session to the end of queue when a transaction starts
         * (depending on isolation mode)
         */
        public void beginAction(Session session, Statement cs)
        {

            lock (liveTransactionTimestamps)
            {
                session.actionTimestamp = nextChangeTimestamp();

                if (!session.isTransaction)
                {
                    session.transactionTimestamp = session.actionTimestamp;
                    session.isTransaction = true;

                    liveTransactionTimestamps.addLast(session.actionTimestamp);

                    try
                    {
                        if (this.mvcc)
                        {
                            database.logger.writeToLog(
                                session, session.getStartTransactionSQL());
                        }
                    }
                    catch (CoreException ) { }
                }
            }

            if (session.isReadOnly())
            {
                return;
            }

            if (!mvcc)
            {
                if (session.hasLocks())
                {
                    return;
                }

                try
                {

#if !SILVERLIGHT
                _lock.EnterWriteLock();
#else
                    Monitor.Enter(_lock);
#endif

                    bool canProceed = beginActionTPL(session, cs);

                    if (!canProceed)
                    {
                        session.abortTransaction = true;
                    }
                    /* debug 190
                                    if (!canProceed) {
                                        System.out.println("******* cannot start");
                                    }
                    // debug 190 */
                }
                finally
                {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                    Monitor.Exit(_lock);
#endif
                }
            }
        }

        public void endTransactionTPL(Session session)
        {

            int unlockedCount = 0;

            if (session.isReadOnly())
            {
                return;
            }

            unlockTablesTPL(session);

            int waitingCount = session.waitingSessions.size();

            if (waitingCount == 0)
            {
                return;
            }

            for (int i = 0; i < waitingCount; i++)
            {
                Session current = (Session)session.waitingSessions.get(i);

                current.tempUnlocked = false;

                long count = current.latch.getCount();

                if (count == 1)
                {
                    bool canProceed = setWaitedSessionsTPL(current,
                        current.currentStatement);

                    if (!canProceed)
                    {
                        current.abortTransaction = true;
                    }

                    if (current.tempSet.isEmpty())
                    {
                        lockTablesTPL(current, current.currentStatement);

                        current.tempUnlocked = true;

                        unlockedCount++;
                    }
                }
            }

            /* debug 190

                        if (unlockedCount > 1) {
                            System.out.println("*********** multiple unlocked");
                        }
            // debug 190 */
            for (int i = 0; i < waitingCount; i++)
            {
                Session current = (Session)session.waitingSessions.get(i);

                if (!current.tempUnlocked)
                {

                    // this can introduce additional waits for the sessions
                    bool canProceed = setWaitedSessionsTPL(current,
                        current.currentStatement);

                    if (!canProceed)
                    {
                        current.abortTransaction = true;
                    }
                    /* debug 190
                                        if (current.tempSet.isEmpty()) {
                                            System.out.println("*********** additional unlocked");

                                            // shouldn't get here
                                        }
                    // debug 190 */
                }
            }

            for (int i = 0; i < waitingCount; i++)
            {
                Session current = (Session)session.waitingSessions.get(i);

                setWaitingSessionTPL(current);
            }

            session.tempSet.clear();
            session.waitingSessions.clear();
        }

        public bool beginActionTPL(Session session, Statement cs)
        {

            if (session.isReadOnly())
            {
                return true;
            }

            bool canProceed = setWaitedSessionsTPL(session, cs);

            if (canProceed)
            {
                if (session.tempSet.isEmpty())
                {
                    lockTablesTPL(session, cs);

                    // we dont set other sessions that would now be waiting for this one too
                }
                else
                {
                    setWaitingSessionTPL(session);
                }

                return true;
            }

            return false;
        }

        public bool setWaitedSessionsTPL(Session session, Statement cs)
        {

            session.tempSet.clear();

            if (cs == null || session.abortTransaction)
            {
                return true;
            }

            QNameManager.QName[] nameList = cs.getTableNamesForWrite();

            for (int i = 0; i < nameList.Length; i++)
            {
                QNameManager.QName name = nameList[i];

                if (name.schema == SqlInvariants.SYSTEM_SCHEMA_QNAME)
                {
                    continue;
                }

                Session holder = (Session)tableWriteLocks.get(name);

                if (holder != null && holder != session)
                {
                    session.tempSet.add(holder);
                }

                Iterator it = tableReadLocks.get(name);

                while (it.hasNext())
                {
                    holder = (Session)it.next();

                    if (holder != session)
                    {
                        session.tempSet.add(holder);
                    }
                }
            }

            nameList = cs.getTableNamesForRead();

            for (int i = 0; i < nameList.Length; i++)
            {
                QNameManager.QName name = nameList[i];

                if (name.schema == SqlInvariants.SYSTEM_SCHEMA_QNAME)
                {
                    continue;
                }

                Session holder = (Session)tableWriteLocks.get(name);

                if (holder != null && holder != session)
                {
                    session.tempSet.add(holder);
                }
            }

            for (int i = 0; i < session.waitingSessions.size(); i++)
            {
                Session current = (Session)session.waitingSessions.get(i);

                if (session.tempSet.contains(current))
                {
                    session.tempSet.clear();

                    return false;
                }
            }

            return true;
        }

        public void setWaitingSessionTPL(Session session)
        {

            int count = session.tempSet.size();

            for (int i = 0; i < count; i++)
            {
                Session current = (Session)session.tempSet.get(i);

                current.waitingSessions.add(session);
            }

            session.tempSet.clear();
            session.latch.setCount(count);
        }

        public void lockTablesTPL(Session session, Statement cs)
        {

            if (cs == null || session.abortTransaction)
            {
                return;
            }

            QNameManager.QName[] nameList = cs.getTableNamesForWrite();

            for (int i = 0; i < nameList.Length; i++)
            {
                QNameManager.QName name = nameList[i];

                if (name.schema == SqlInvariants.SYSTEM_SCHEMA_QNAME)
                {
                    continue;
                }

                tableWriteLocks.put(name, session);
            }

            nameList = cs.getTableNamesForRead();

            for (int i = 0; i < nameList.Length; i++)
            {
                QNameManager.QName name = nameList[i];

                if (name.schema == SqlInvariants.SYSTEM_SCHEMA_QNAME)
                {
                    continue;
                }

                tableReadLocks.put(name, session);
            }
        }

        public void unlockTablesTPL(Session session)
        {

            Iterator it = tableWriteLocks.values().iterator();

            while (it.hasNext())
            {
                Session s = (Session)it.next();

                if (s == session)
                {
                    it.setValue(null);
                }
            }

            it = tableReadLocks.values().iterator();

            while (it.hasNext())
            {
                Session s = (Session)it.next();

                if (s == session)
                {
                    it.remove();
                }
            }
        }

        /**
         * remove session from queue when a transaction ends
         * and expire any committed transactions
         * that are no longer required. remove transactions ended before the first
         * timestamp in liveTransactionsSession queue
         */
        public void endTransaction(Session session)
        {

            try
            {
#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
                Monitor.Enter(_lock);
#endif

                long timestamp = session.transactionTimestamp;

                lock (liveTransactionTimestamps)
                {
                    session.isTransaction = false;

                    int index = liveTransactionTimestamps.indexOf(timestamp);

                    liveTransactionTimestamps.remove(index);
                }

                mergeExpiredTransactions(session);
            }
            finally
            {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
            }
        }

        public long getFirstLiveTransactionTimestamp()
        {

            lock (liveTransactionTimestamps)
            {
                if (liveTransactionTimestamps.isEmpty())
                {
                    return long.MaxValue;
                }

                return liveTransactionTimestamps.get(0);
            }
        }

        // functional unit - list actions and translate id's

        /**
         * Return an array of all row actions sorted by System Change No.
         */
        public RowAction[] getRowActionList()
        {

            try
            {
#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
                Monitor.Enter(_lock);
#endif

                Session[] sessions = database.sessionManager.getAllSessions();
                int[] tIndex = new int[sessions.Length];
                RowAction[] rowActions;
                int rowActionCount = 0;

                {
                    int actioncount = 0;

                    for (int i = 0; i < sessions.Length; i++)
                    {
                        actioncount += sessions[i].getTransactionSize();
                    }

                    rowActions = new RowAction[actioncount];
                }

                while (true)
                {
                    bool found = false;
                    long minChangeNo = long.MaxValue;
                    int sessionIndex = 0;

                    // find the lowest available SCN across all sessions
                    for (int i = 0; i < sessions.Length; i++)
                    {
                        int tSize = sessions[i].getTransactionSize();

                        if (tIndex[i] < tSize)
                        {
                            RowAction current =
                                (RowAction)sessions[i].rowActionList.get(
                                    tIndex[i]);

                            if (current.actionTimestamp < minChangeNo)
                            {
                                minChangeNo = current.actionTimestamp;
                                sessionIndex = i;
                            }

                            found = true;
                        }
                    }

                    if (!found)
                    {
                        break;
                    }

                    EfzArrayList currentList =
                        sessions[sessionIndex].rowActionList;

                    for (; tIndex[sessionIndex] < currentList.size(); )
                    {
                        RowAction current =
                            (RowAction)currentList.get(tIndex[sessionIndex]);

                        // if the next change no is in this session, continue adding
                        if (current.actionTimestamp == minChangeNo + 1)
                        {
                            minChangeNo++;
                        }

                        if (current.actionTimestamp == minChangeNo)
                        {
                            rowActions[rowActionCount++] = current;

                            tIndex[sessionIndex]++;
                        }
                        else
                        {
                            break;
                        }
                    }
                }

                return rowActions;
            }
            finally
            {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
            }
        }

        /**
         * Return a lookup of all row ids for cached tables in transactions.
         * For auto-defrag, as currently there will be no RowAction entries
         * at the time of defrag.
         */
        public DoubleIntIndex getTransactionIDList()
        {

#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
            Monitor.Enter(_lock);
#endif

            try
            {
                int size = rowActionMap.size();
                DoubleIntIndex lookup = new DoubleIntIndex(size, false);

                lookup.setKeysSearchTarget();

                Iterator it = this.rowActionMap.keySet().iterator();

                for (; it.hasNext(); )
                {
                    lookup.addUnique(it.nextInt(), 0);
                }

                return lookup;
            }
            finally
            {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
            }
        }

        /**
         * Convert row ID's for cached table rows in transactions
         */
        public void convertTransactionIDs(DoubleIntIndex lookup)
        {

#if !SILVERLIGHT
            _lock.EnterWriteLock();
#else
            Monitor.Enter(_lock);
#endif

            try
            {
                RowAction[] list = new RowAction[rowActionMap.size()];
                Iterator it = this.rowActionMap.values().iterator();

                for (int i = 0; it.hasNext(); i++)
                {
                    list[i] = (RowAction)it.next();
                }

                rowActionMap.clear();

                for (int i = 0; i < list.Length; i++)
                {
                    int pos = lookup.lookupFirstEqual(list[i].getPos());

                    list[i].setPos(pos);
                    rowActionMap.put(pos, list[i]);
                }
            }
            finally
            {
#if !SILVERLIGHT
                _lock.ExitWriteLock();
#else
                Monitor.Exit(_lock);
#endif
            }
        }
    }
}
